<?php
// 消费主队列，处理任务后写入回调队列
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
$redis->select(1);

$mainStream = 'task:main:queue';
$groupName = 'main_group';
$consumerName = 'worker1';

while (true) {
    $tasks = $redis->xReadGroup($groupName, $consumerName, [$mainStream => '>'], 1, 1000);
    if (!empty($tasks[$mainStream])) {
        foreach ($tasks[$mainStream] as $msgId => $fields) {
            echo "处理任务: $msgId\n";
            $payload = json_decode($fields['payload'], true);
            // 简单处理：a+b
            $result = $payload['a'] + $payload['b'];

            // 回调
            $callbackStream = $fields['callback_stream'];
            $callbackData = [
                'task_id' => $fields['task_id'],
                'result' => $result
            ];
            $redis->xAdd($callbackStream, '*', $callbackData);
            echo "已写入回调队列: $callbackStream\n";

            // ack 确认
            $redis->xAck($mainStream, $groupName, [$msgId]);
            echo "已确认任务: $msgId\n";
            
            // 删除已处理的任务
            $deletedCount = $redis->xDel($mainStream, [$msgId]);
            echo "已删除任务: $msgId (删除数量: $deletedCount)\n";
        }
    }
    usleep(500000); // 0.5秒
} 